package com.example.mq.factory;

import com.example.mq.annotation.MQListener;
import com.example.mq.config.SysMQConfig;
import com.example.mq.dto.MQListenerDto;
import com.example.mq.listener.SysMQMessageListener;
import com.example.mq.po.SysMqReceiveLog;
import com.example.mq.service.SysMqReceiveLogService;
import com.example.server.service.SysServerService;
import com.example.sync.handler.impl.SysMQMessageRetryHandler;
import com.example.sync.handler.impl.wrap.SysMQRetryMessage;
import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/*
 * @className: SysRabbitMQFactory
 * @author: chentao
 * @date: 2025/3/25 下午3:14
 * @Version: 1.0
 * @description:
 */
public class SysRabbitMQFactory extends SysAbstractMQFactory {

    Logger logger = LoggerFactory.getLogger(getClass());

    private SysMQConfig sysMQConfig;

    private ApplicationContext applicationContext;

    private SysServerService sysServerService;

    private SysMQMessageRetryHandler sysMQMessageRetryHandler;

    private SysMqReceiveLogService sysMqReceiveLogService;

    private ExecutorService executorService;

    private ConnectionFactory connectionFactory;

    private Connection connection;

    private Map<Integer, Channel> channelMaps = new ConcurrentHashMap<>();

    @Override
    public void init(SysMQConfig sysMQConfig, ApplicationContext applicationContext, SysServerService sysServerService) {
        try {
            this.sysMQConfig = sysMQConfig;
            this.applicationContext = applicationContext;
            this.sysServerService = sysServerService;
            this.sysMQMessageRetryHandler = applicationContext.getBean(SysMQMessageRetryHandler.class);
            this.sysMqReceiveLogService = applicationContext.getBean(SysMqReceiveLogService.class);
            this.executorService = Executors.newCachedThreadPool();
            this.connectionFactory = new ConnectionFactory();
            this.connectionFactory.setHost(sysMQConfig.getHost());
            this.connectionFactory.setPort(sysMQConfig.getPort());
            this.connectionFactory.setUsername(sysMQConfig.getUsername());
            this.connectionFactory.setPassword(sysMQConfig.getPassword());
            this.connection = this.connectionFactory.newConnection();
        } catch (Exception e) {
            logger.warn("RabbitMQ 初始化失败: {}", e.getMessage());
        }
    }

    @Override
    public void preValidate(Map<String, MQListenerDto> mqListenerDtoMap) throws Exception {
        for (MQListenerDto mqListenerDto : mqListenerDtoMap.values()) {
            MQListener mqListener = mqListenerDto.getMqListener();
            if (StringUtils.isEmpty(mqListener.queue())) {
                throw new Exception("bean: " + mqListenerDto.getBeanName() + " 对应queue 不可以为空.");
            }
            if (StringUtils.isEmpty(mqListener.exchange())) {
                throw new Exception("bean: " + mqListenerDto.getBeanName() + " 对应exchange 不可以为空.");
            }
            if (StringUtils.isEmpty(mqListener.routingKey())) {
                throw new Exception("bean: " + mqListenerDto.getBeanName() + " 对应routingKey 不可以为空.");
            }
        }
    }

    @Override
    public void start(Map<String, MQListenerDto> mqListenerDtoMap) {
        this.executorService.execute(new Runnable() {
            @Override
            public void run() {
                for (String beanName : mqListenerDtoMap.keySet()) {
                    try {
                        MQListenerDto mqListenerDto = mqListenerDtoMap.get(beanName);
                        MQListener mqListener = mqListenerDto.getMqListener();
                        Channel channel = connection.createChannel();
                        String queueName = mqListener.queue();
                        if (mqListener.isTopic()) {
                            queueName = queueName + sysServerService.getCurrentServerId();
                            channel.queueDeclare(queueName, false, false, false, null);
                            channel.exchangeDeclare(mqListener.exchange(), BuiltinExchangeType.FANOUT, true);
                        } else {
                            channel.queueDeclare(queueName, false, false, false, null);
                            channel.exchangeDeclare(mqListener.exchange(), BuiltinExchangeType.DIRECT, true);
                        }
                        channel.queueBind(queueName, mqListener.exchange(), mqListener.routingKey());
                        channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
                            @Override
                            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                                executorService.execute(new RabbitMQMessageHandleThead(mqListenerDto, channel, envelope, body));
                            }
                        });
                        channelMaps.put(channel.getChannelNumber(), channel);
                    } catch (Exception e) {
                        logger.warn("rabbitmq listener init error: " + e.getMessage());
                    }
                }

            }
        });
    }

    @Override
    public void stop() throws Exception {
        for (Channel channel : this.channelMaps.values()) {
            if (channel != null && channel.isOpen()) {
                channel.close();
            }
        }
        this.connection.close();
        this.executorService.shutdownNow();
    }

    //mq消息处理线程
    class RabbitMQMessageHandleThead implements Runnable {

        private MQListenerDto mqListenerDto;

        private Channel channel;

        private Envelope envelope;

        private byte[] bytes;

        public RabbitMQMessageHandleThead(MQListenerDto mqListenerDto, Channel channel, Envelope envelope, byte[] bytes) {
            this.mqListenerDto = mqListenerDto;
            this.channel = channel;
            this.envelope = envelope;
            this.bytes = bytes;
        }

        @SneakyThrows
        @Override
        public void run() {
            String message = new String(bytes, StandardCharsets.UTF_8);
            MQListener mqListener = mqListenerDto.getMqListener();
            SysMqReceiveLog sysMqReceiveLog = null;
            SysMQMessageListener sysMQMessageListener = null;
            try {
                sysMqReceiveLog = new SysMqReceiveLog(SysRabbitMQFactory.class.getName(), mqListener.exchange(), mqListener.queue(), mqListener.routingKey(), mqListenerDto.getBeanName(), sysServerService.getCurrentServerId(), message);
                sysMQMessageListener = (SysMQMessageListener) applicationContext.getBean(mqListenerDto.getBeanName());
                sysMQMessageListener.onMessage(message);
            } catch (Exception e) {
                sysMqReceiveLog.setStatus(false);
                SysMQRetryMessage sysMQRetryMessage = new SysMQRetryMessage(mqListener.queue(), mqListenerDto.getBeanName(), message);
                sysMQMessageRetryHandler.publish(SysRabbitMQFactory.class.getName(), 0L, "mq.message.retry", sysMQRetryMessage.getClass().getName(), sysMQRetryMessage);
                logger.warn("mq bean : " + sysMQMessageListener.getClass().getName() + " 消费消息失败: " + e.getMessage());
            } finally {
                sysMqReceiveLog.recalculateFields();
                sysMqReceiveLogService.save(sysMqReceiveLog);
            }
        }
    }

    @Override
    public void publish(String exchange, String routingKey, String content) throws Exception {
        super.publish(exchange, routingKey, content);
        Channel channel = null;
        try {
            channel = connection.createChannel();
            channel.basicPublish(exchange, routingKey, null, content.getBytes(StandardCharsets.UTF_8));
        } catch (Exception e) {
            logger.warn("RabbitMQ send error : {}", e.getMessage());
            throw e;
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (Exception e) {
                    logger.warn("close channel error : {}", e.getMessage());
                }
            }
        }
    }

    public Connection getConnection() {
        return connection;
    }
}
